package com.lagou.client;

import com.lagou.handler.UserClientHandler;
import com.lagou.pojo.RpcRequest;
import com.lagou.pojo.RpcResponse;
import com.lagou.serializer.JsonSerializer;
import com.lagou.serializer.RpcDecoder;
import com.lagou.serializer.RpcEncoder;
import com.lagou.zk.CommonNode;
import com.lagou.zk.ZookeeperClient;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.List;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * 消费者
 */
@Component
public class RPCConsumer implements PathChildrenCacheListener {

    //1.创建一个线程池对象  -- 它要处理我们自定义事件
    private ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    @Autowired
    private JsonSerializer serializer;

    private Map<String, UserClientHandler> handlerMap = new HashMap<>();

    public void initClient(String host) throws Exception {
        bootStrapConnection(host);
    }

    // 编写方法,初始化客户端  ( 创建连接池  bootStrap  设置bootstrap  连接服务器)
    public void initClient() throws Exception {
        CuratorFramework client = ZookeeperClient.getClient();
        // 返回的是子节点id
        List<String> serverIdNodesList = client.getChildren().forPath(CommonNode.serverNode);
        // key值为  ip:port
        for (String hostNode : serverIdNodesList) {
            String host = new String(client.getData().forPath(CommonNode.serverNode + "/" + hostNode));
            // 如果handlerMap中不存在该host，说明未绑定，直接绑定该端口
            bootStrapConnection(host);
        }
    }

    /**
     * 用于通过host连接服务端
     * @param host
     * @throws Exception
     */
    private void bootStrapConnection(String host) throws Exception {

        // 如果已经存在该连接，则直接返回
        if (handlerMap.size() > 0 && handlerMap.containsKey(host)) {
            return;
        }

        //1)初始化UserClientHandler
        UserClientHandler userClientHandler = new UserClientHandler();
        //2)创建连接池对象
        EventLoopGroup group = new NioEventLoopGroup();
        //3)创建客户端的引导对象
        Bootstrap bootstrap = new Bootstrap();
        //4)配置启动引导对象
        bootstrap.group(group)
                //设置通道为NIO
                .channel(NioSocketChannel.class)
                //设置请求协议为TCP
                .option(ChannelOption.TCP_NODELAY, true)
                //监听channel 并初始化
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        //获取ChannelPipeline
                        ChannelPipeline pipeline = ch.pipeline();
                        //设置编码
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, serializer));
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, serializer));
                        //添加自定义事件处理器
                        pipeline.addLast(userClientHandler);
                    }
                });

        userClientHandler.setHost(host);
        handlerMap.put(host, userClientHandler);
        String[] split = host.split(":");
        // 连接服务端
        bootstrap.connect(split[0], Integer.valueOf(split[1])).sync();
    }

    public void removeHandler(String hosts){
        if (handlerMap.containsKey(hosts)) {
            handlerMap.remove(hosts);
            System.out.println(hosts + "节点已下线");
        }
    }

    /**
     * 获取ResponseTime最短的代理对象，如果该Handler中的代理对象尚未生成，则直接创建该对象
     * @param serviceClass
     * @return
     */
    public Object getProxy(Class<?> serviceClass){
        UserClientHandler userClientHandler = null;
        Long responseTime = null;
        String host = null;
        for (Map.Entry<String, UserClientHandler> entry : handlerMap.entrySet()) {
            UserClientHandler tempHandler = entry.getValue();
            String tempHost = entry.getKey();

            // 如果Handler中的代理对象不存在，则创建代理对象并保存在Handler中
            Object proxyObject = tempHandler.getProxyObject();
            if (proxyObject == null) {
                Object tempObj = createProxy(serviceClass, tempHost);
                if (tempObj != null) {
                    tempHandler.setProxyObject(tempObj);
                }
            }
            System.out.println( tempHost + "服务器最后一次请求响应时间为：" + tempHandler.getResponseTime());

            // 当第一次进入循环或者时间出现更小值时，使用时间更短的Handler
            if (responseTime == null || (responseTime - tempHandler.getResponseTime() > 0)) {
                responseTime = tempHandler.getResponseTime();
                userClientHandler = tempHandler;
                host = tempHost;
            }
        }
        System.out.println("当前正在使用的服务器为：" + host);
        System.out.println("当前正在使用的服务器的响应时间为：" + responseTime);
        return userClientHandler.getProxyObject();
    }

    public Object createProxy(Class<?> serviceClass, String host){
        // 如果该host存在，并且尚未生成对象，则新建代理类
        if (handlerMap.containsKey(host) && handlerMap.get(host).getProxyObject() == null) {
            return createProxy(serviceClass, handlerMap.get(host));
        }
        return null;
    }

    //4.编写一个方法,使用JDK的动态代理创建对象
    private Object createProxy(Class<?> serviceClass, UserClientHandler handler) {
        // Thread.currentThread()
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    public Object invoke(Object o, Method method, Object[] objects) throws Throwable {

                        // 初始化RpcRequest对象，并放入信道
                        RpcRequest request = new RpcRequest();
                        request.setRequestId(UUID.randomUUID().toString());
                        request.setClassName(serviceClass.getName());
                        request.setMethodName(method.getName());
                        request.setParameters(objects);
                        request.setParameterTypes(method.getParameterTypes());

                        //2)给UserClientHandler 设置param参数
                        handler.setParam(request);

                        //3).使用线程池,开启一个线程处理处理call() 写操作,并返回结果
                        Object result = executorService.submit(handler).get();

                        //4)return 结果
                        return result;
                    }
                });
    }

    @Override
    public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
        PathChildrenCacheEvent.Type type = event.getType();
        if (PathChildrenCacheEvent.Type.CONNECTION_LOST.equals(type) || PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED.equals(type)
        || PathChildrenCacheEvent.Type.CONNECTION_SUSPENDED.equals(type)) {
            System.out.println("发现连接事件：" + type);
            return;
        }

        String data = new String(event.getData().getData());

        System.out.println("变动节点type：" + type);
        System.out.println("变动节点Path：" + event.getData().getPath());
        System.out.println("变动节点data：" + data);

        if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(type)) {
            // 子节点添加或者修改时，绑定新增或修改后的端口
            System.out.println("新增Host：" + data);
            initClient(data);
        } else if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(type)) {
            // 子节点删除时，解除端口绑定（其实这一步在Handler监听里面就已经实现了，在这里只删除对象）
            System.out.println("删除Host：" + data);
            removeHandler(data);
        }
    }
}
